package com.yusc.write_to_hdfs;

import com.google.gson.Gson;
import com.opencsv.CSVReader;
import com.opencsv.CSVReaderBuilder;
import com.opencsv.exceptions.CsvValidationException;
import com.yusc.pojo.ALlBasicInfo;
import com.yusc.utils.WriteHdfsHandler;

import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author Yusc
 * @Date 2023/11/13 23:18
 * @Version
 * @Description 基于openScv读取scv文件并写入hdfs
 **/

public class WriteToHdfs {
    private static final Gson gson = new Gson();
    private static final List<ALlBasicInfo> infoList = new ArrayList<>();

    public static void main(String[] args) throws IOException, CsvValidationException {
        String fileUrl = "src/main/resources/data/air_data.csv";
        InputStreamReader reader = new InputStreamReader(Files.newInputStream(Paths.get(fileUrl)), StandardCharsets.UTF_8);

        CSVReader csvReader = new CSVReaderBuilder(reader).build();

        String airDataUrl = "/air_data";
        String defaultFS = "hdfs://master:9000";
        WriteHdfsHandler writeHdfsHandler = new WriteHdfsHandler(airDataUrl, defaultFS);

        String[] str = csvReader.readNext();    // 数据读取 ---> 第一行不需要
        int index = 0;  // 用来记录行数

        while (true) {
            str = csvReader.readNext();
            if (str == null) {
                break;
            }

            infoList.add(toBean(str));
            index++;
            if (index == 10000) {
                // 将数据写入到HDFS
                writeHdfsHandler.writeCanDataToHdfs(infoList);
                System.out.println("写入了：" + index + "条数据");
                infoList.clear();   //将列表清空
                index = 0;
            }
        }
        // 关闭连接
        writeHdfsHandler.close();
        csvReader.close();
    }

    private static ALlBasicInfo toBean(String[] strS) {
        ALlBasicInfo newInfo = new ALlBasicInfo();

        newInfo.setMemberNo(strS[0]);

        try {
            newInfo.setFfpDate(LocalDate.parse(strS[1], DateTimeFormatter.ofPattern("yyyy/M/d")));
        } catch (Exception e) {
            newInfo.setFfpDate(null);
        }
        try {
            newInfo.setFirstFlightDate(LocalDate.parse(strS[2], DateTimeFormatter.ofPattern("yyyy/M/d")));
        } catch (Exception e) {
            newInfo.setFirstFlightDate(null);
        }
        newInfo.setGender(strS[3]);

        try {
            newInfo.setFfpTier(Integer.valueOf(strS[4]));
        } catch (Exception e) {
            newInfo.setFfpTier(null);
        }

        if ("".equals(strS[5])) {
            newInfo.setWorkCity(null);
        } else {
            newInfo.setWorkCity(strS[5]);
        }
        if ("".equals(strS[6])) {
            newInfo.setWorkProvince(null);
        } else {
            newInfo.setWorkProvince(strS[6]);
        }
        if ("".equals(strS[7])) {
            newInfo.setWorkCountry(null);
        } else {
            newInfo.setWorkCountry(strS[7]);
        }

        try {
            newInfo.setAge(Integer.valueOf(strS[8]));
        } catch (Exception e) {
            newInfo.setAge(null);
        }
        try {
            newInfo.setFlightCount(Integer.valueOf(strS[9]));
        } catch (Exception e) {
            newInfo.setFlightCount(null);
        }
        try {
            newInfo.setAvgDiscount(Double.valueOf(strS[10]));
        } catch (Exception e) {
            newInfo.setAvgDiscount(null);
        }
        try {
            newInfo.setSegKmSum(Long.valueOf(strS[11]));
        } catch (Exception e) {
            newInfo.setSegKmSum(null);
        }
        try {
            newInfo.setLastFlightDate(LocalDate.parse(strS[12], DateTimeFormatter.ofPattern("yyyy/M/d")));
        } catch (Exception e) {
            newInfo.setLastFlightDate(null);
        }
        try {
            newInfo.setAvgInterval(Double.valueOf(strS[13]));
        } catch (Exception e) {
            newInfo.setAvgInterval(null);
        }
        try {
            newInfo.setMaxInterval(Integer.valueOf(strS[14]));
        } catch (Exception e) {
            newInfo.setMaxInterval(null);
        }
        try {
            newInfo.setP1yFlightCount(Integer.valueOf(strS[15]));
        } catch (Exception e) {
            newInfo.setP1yFlightCount(null);
        }
        try {
            newInfo.setL1yFlightCount(Integer.valueOf(strS[16]));
        } catch (Exception e) {
            newInfo.setL1yFlightCount(null);
        }
        try {
            newInfo.setSumYr1(Long.valueOf(strS[17]));
        } catch (Exception e) {
            newInfo.setSumYr1(null);
        }
        try {
            newInfo.setSumYr2(Long.valueOf(strS[18]));
        } catch (Exception e) {
            newInfo.setSumYr2(null);
        }
        try {
            newInfo.setEpSum(Long.valueOf(strS[19]));
        } catch (Exception e) {
            newInfo.setEpSum(null);
        }
        try {
            newInfo.setBpSum(Long.valueOf(strS[20]));
        } catch (Exception e) {
            newInfo.setBpSum(null);
        }
        try {
            newInfo.setPointsSum(Long.valueOf(strS[21]));
        } catch (Exception e) {
            newInfo.setPointsSum(null);
        }
        try {
            newInfo.setExchangeCount(Integer.valueOf(strS[22]));
        } catch (Exception e) {
            newInfo.setExchangeCount(null);
        }
        try {
            newInfo.setP1yBpSum(Long.valueOf(strS[23]));
        } catch (Exception e) {
            newInfo.setP1yBpSum(null);
        }
        try {
            newInfo.setL1yBpSum(Long.valueOf(strS[24]));
        } catch (Exception e) {
            newInfo.setL1yBpSum(null);
        }
        try {
            newInfo.setL1yPointsSum(Long.valueOf(strS[25]));
        } catch (Exception e) {
            newInfo.setL1yPointsSum(null);
        }
        try {
            newInfo.setEliAddPointSum(Long.valueOf(strS[26]));
        } catch (Exception e) {
            newInfo.setEliAddPointSum(null);
        }
        try {
            newInfo.setPointNotFlight(Integer.valueOf(strS[27]));
        } catch (Exception e) {
            newInfo.setPointNotFlight(null);
        }
        try {
            newInfo.setLstToEnd(Double.valueOf(strS[28]));
        } catch (Exception e) {
            newInfo.setLstToEnd(null);
        }
        try {
            newInfo.setLoadTime(LocalDate.parse(strS[29], DateTimeFormatter.ofPattern("yyyy/M/d")));
        } catch (Exception e) {
            newInfo.setLoadTime(null);
        }
        return newInfo;
    }
}
